package com.luoqifei.yomob.realtime.analysis.kafka.stream;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lqf.common.Constants;
import com.lqf.common.ParamNames;
import com.lqf.common.RecordTypeEnum;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class DAUCountKStream {
    public static final Logger LOG = LoggerFactory.getLogger(DAUCountKStream.class);
    private Properties config;
    private KafkaStreams streams;
    private KStreamBuilder builder = new KStreamBuilder();

    public DAUCountKStream() {
        KStream<String, String> dauRecord = builder.stream(Constants.DAU_SOURCE_TOPIC);

        //no key ,or the key is not appid
        KTable<String, Long> dauCounts = dauRecord.filter((key, value) -> value.length() > 0)
                .filter((key, value) -> {
                    try {
                        JSONObject jsonObject = JSON.parseObject(value);
                        boolean flag = jsonObject.containsKey(ParamNames.COUNTER_RECORD_APPID)
                                && jsonObject.get(ParamNames.COUNTER_RECORD_APPID).toString().length() > 0
                                && jsonObject.containsKey(ParamNames.COUNTER_RECORD_COUNTER_ID)
                                && jsonObject.get(ParamNames.COUNTER_RECORD_COUNTER_ID).toString()
                                .equals(RecordTypeEnum.ACTIVE.toString());
                        if (!flag) {
                            LOG.warn("get an unknown record, key={}, value={}", key, value);
                        }
                        return flag;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return false;
                    }
                })
                .groupBy((key, value) -> JSON.parseObject(value).get(ParamNames.COUNTER_RECORD_APPID).toString())
                .count(Constants.DAU_COUNT_STATE_STORE);
        dauCounts.to(Serdes.String(), Serdes.Long(), Constants.DAU_COUNT_TOPIC);

    }

    /**
     * default k-v Type is String
     *
     * @param kafkaServerAddress
     */
    private void initDefaultConfig(String kafkaServerAddress) {
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, Constants.DAU_APPLICATION_ID);
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServerAddress);
        config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);

    }

    public void start() {
        streams = new KafkaStreams(builder, config);
        streams.start();
    }
}
